package com.pass.two;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.data.ConnectedEdge;
import com.uf.UnionFind;
import com.utils.CommonUtils;

/**
 * Pass two reducer, input: <-1, Edge Edge Edge...>, performs UF on ALL edges,
 * and output UF result to file.
 */
public class PassTwoReducer extends
		Reducer<IntWritable, Text, IntWritable, Text> {
	@Override
	protected void reduce(IntWritable key, Iterable<Text> values,
			Context context) throws IOException, InterruptedException {
		Set<ConnectedEdge> edges = new HashSet<ConnectedEdge>();
		for (Text value : values) {
			edges.addAll(CommonUtils.getEdgesFromString(value.toString()));
		}
		UnionFind uf = new UnionFind(edges);
		Set<ConnectedEdge> ufEdges = uf.getUFEdges();
		for (ConnectedEdge edge : ufEdges) {
			context.write(key, new Text(edge.toString()));
		}
	}

}
